package cn.learn.config;

import org.elasticsearch.action.bulk.*;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * @author huangyezhan
 * @version 1.0
 * @date 2020年06月30日15:02
 */
@Configuration
public class ElasticsearchBulkProcessorConfig {
    private static final Logger logger = LoggerFactory.getLogger(ElasticsearchBulkProcessorConfig.class);

    private final RestHighLevelClient restHighLevelClient;

    public ElasticsearchBulkProcessorConfig(RestHighLevelClient restHighLevelClient) {
        this.restHighLevelClient = restHighLevelClient;
    }


    /**
     * 批量处理器（异步）
     * 批量处理会有数据丢失问题
     *
     * @return 获取对象
     */
    @Bean
    public BulkProcessor bulkProcessor() {
        //批量处理监听器
        BulkProcessor.Listener listener = new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {

            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                if (response.hasFailures()) {
                    for (BulkItemResponse bulkItemResponse : response) {
                        //遍历失败的
                        if (bulkItemResponse.isFailed()) {
                            BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                            logger.info("失败的id：[{}]；失败的原因：{}", bulkItemResponse.getId(), failure.getMessage());
                        }
                    }
                }
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                logger.info("异常原因：", failure.getCause());
            }
        };

        BulkProcessor.Builder builder = BulkProcessor.builder((request, bulkListener) ->
                restHighLevelClient.bulkAsync(request, RequestOptions.DEFAULT, bulkListener), listener);

        //默认是1000
        builder.setBulkActions(1000);
        //默认是5M
        builder.setBulkSize(new ByteSizeValue(5L, ByteSizeUnit.MB));
        //设置允许执行的并发请求数（默认为1，使用0仅允许执行单个请求）
        //不是集群,启用多个线程，会丢失数据，启用得越多，丢失得越多，es的IO瓶颈
        builder.setConcurrentRequests(1);
        builder.setFlushInterval(TimeValue.timeValueSeconds(10L));
        builder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1L), 3));

        return builder.build();
    }


}
